/**
 * Date:2022年1月6日 上午10:58:20
 * Copyright (c) 2022, www.bwbroad.com All Rights Reserved
 */

package com.example.action;

import java.util.Optional;
import java.util.function.Supplier;

import org.springframework.data.repository.CrudRepository;
import org.springframework.kafka.support.Acknowledgment;

import com.alibaba.fastjson.JSON;
import com.example.common.utils.BeanUtilsX;
import com.example.common.utils.JavaBeanUtils;
import com.example.vo.OggMessageMap;

/**
 * Description: TODO <br/>
 * Date:2022年1月6日 上午10:58:20 <br/>
 * 
 * @author xuejianxin
 */
public abstract class BaseKafkaConsumer {
	private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(KafkaConsumer2.class);

	protected <T, K> void consumer(String message, K key, String topic, Acknowledgment ack, CrudRepository<T, K> dao,
			Supplier<T> newEntity) {
		log.info(">>>>>>>topic={},key={}, consume message: {}", topic, key, message);
		OggMessageMap msgObj = JSON.parseObject(message, OggMessageMap.class);
		// key 就是数据的主键
		Optional<T> optEn = dao.findById(key);
		T en = optEn.orElseGet(newEntity);

		try {
			switch (msgObj.getOpType()) {
			case I:
				// 把map中的值合并到en
				BeanUtilsX.copyProperties(en, JavaBeanUtils.underlineToCamel(msgObj.getAfter()));
				// 有则更新无则插入
				dao.save(en);
				break;
			case U:
				if (optEn.isPresent()) {// 存在则更新,不存在也不插入了,因为作为更新的数据很多字段为null,在插入的时候无法通过null 检查
					// 把map中的值合并到en
					BeanUtilsX.copyProperties(en, JavaBeanUtils.underlineToCamel(msgObj.getAfter()));
					// 有则更新无则插入
					dao.save(en);
				}
				break;
			case D:
				if (optEn.isPresent()) {
					dao.deleteById(key);
				}
				break;

			default:
				break;
			}
			// 应答
			ack.acknowledge();
		} catch (Exception e) {
			log.error(topic + ",id=" + key, e);
		}
	}
}
